/*
 *  Copyright 2008 Martin Fryč (martin@fryc.eu).
 * 
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 * 
 *       http://www.apache.org/licenses/LICENSE-2.0
 * 
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *  under the License.
 */
package eu.fryc.martin.micromq.storage.impl;

import eu.fryc.martin.micromq.storage.DestinationManager;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;

/**
 * Transactional implementation of MicroMQ storage engine.
 * 
 * @author mfryc
 */
public class MicroMQTxStorageImpl extends MicroMQStorageAbstractImpl {

	/**
	 * Storage for uncommited incomming messages. Key is conenction id and value
	 * is uncommited message.
	 */
	private Map<String, Queue<Message>> incommingMessages;
	/**
	 * Storage for uncommited outcomming messages. Key is connection id and
	 * value is uncommited message.
	 */
	private Map<String, Queue<Message>> outcommingMessages;

	/**
	 * Put message to destination. Method put message to connection queue of
	 * messages waiting for commit.
	 * 
	 * @param connectionId
	 *            connection id
	 * @param message
	 *            message
	 * 
	 * @throws javax.jms.JMSException
	 *             if put fail
	 * @throws javax.jms.InvalidDestinationException
	 *             if no such destination exists
	 */
	@Override
	public void put(String connectionId, Message message) throws JMSException, InvalidDestinationException {
		checkConnectionId(connectionId);
		if (message == null) {
			throw new JMSException("Cannot put null message.");
		}

		// check if destination exists
		Destination messageDestination = message.getJMSDestination();
		if (!getDestinationManager(messageDestination).exists(messageDestination)) {
			throw new InvalidDestinationException("Destination (" + messageDestination.toString()
					+ ") does not exists.");
		}

		Queue<Message> incommingMessageQueue = null;

		// get connection queue
		synchronized (incommingMessages) {
			incommingMessageQueue = incommingMessages.get(connectionId);
			if (incommingMessageQueue == null) {
				incommingMessageQueue = new LinkedList<Message>();
				incommingMessages.put(connectionId, incommingMessageQueue);
			}
		}

		// add message to queue
		synchronized (incommingMessageQueue) {
			incommingMessageQueue.offer(message);
		}
	}

	/**
	 * Get message from destination. Put message on connection received queue of
	 * uncommited messages.
	 * 
	 * @param destination
	 * @param connectionId
	 * @param timeout
	 *            timeout of get method (0 - no wait, negative - infinite wait,
	 *            other - milis wait time)
	 * 
	 * @return message or null if timeout occurs and destination does not
	 *         contain any message
	 * 
	 * @throws javax.jms.JMSException
	 *             if get fail
	 * @throws javax.jms.InvalidDestinationException
	 *             if no such destination exists
	 */
	@Override
	public Message get(Destination destination, String connectionId, long timeout) throws JMSException,
			InvalidDestinationException {
		if (destination == null) {
			throw new JMSException("Cannot get message from null destination.");
		}
		checkConnectionId(connectionId);

		// prepare timeout
		if (timeout == -1) {
			timeout = Long.MAX_VALUE;
		}
		DestinationManager destinationManager = getDestinationManager(destination);
		Message message = null;
		long receiveStart = System.currentTimeMillis();
		// get message in timeout
		do {
			synchronized (destinationManager) {
				message = destinationManager.get(destination);
			}
			Thread.yield();
		} while (message != null || receiveStart + timeout > System.currentTimeMillis());

		// put message to outcomming queue
		Queue<Message> outcommingMessageQueue = null;

		// get connection queue
		synchronized (outcommingMessages) {
			outcommingMessageQueue = outcommingMessages.get(connectionId);
			if (outcommingMessageQueue == null) {
				outcommingMessageQueue = new LinkedList<Message>();
				outcommingMessages.put(connectionId, outcommingMessageQueue);
			}
		}

		// add message to queue
		synchronized (outcommingMessageQueue) {
			outcommingMessageQueue.offer(message);
		}

		return message;
	}

	/**
	 * Commit messages included in transaction. Behaviour of thid method depends
	 * on session configuration.
	 * 
	 * @param connectionId
	 *            connection id
	 * 
	 * @throws javax.jms.JMSException
	 *             if commit fail
	 */
	@Override
	public void commit(String connectionId) throws JMSException {
		checkConnectionId(connectionId);

		Queue<Message> messageQueue = null;

		// move messages from incomming queue to main queue
		synchronized (incommingMessages) {
			messageQueue = incommingMessages.get(connectionId);
		}
		if (messageQueue != null) {
			synchronized (messageQueue) {
				for (Message message : messageQueue) {
					getDestinationManager(message.getJMSDestination()).put(message.getJMSDestination(), message);
				}
				messageQueue.clear();
			}
		}
		messageQueue = null;

		// delete messages from outcomming queue
		synchronized (outcommingMessages) {
			messageQueue = outcommingMessages.get(connectionId);
		}
		if (messageQueue != null) {
			synchronized (messageQueue) {
				messageQueue.clear();
			}
		}
	}

	/**
	 * Rollback messages included in this transaction. Behaviour of thid method
	 * depends on session configuration.
	 * 
	 * @param connectionId
	 *            connection id
	 * 
	 * @throws javax.jms.JMSException
	 *             if commit fail
	 */
	@Override
	public void rollback(String connectionId) throws JMSException {
		checkConnectionId(connectionId);

		Queue<Message> messageQueue = null;

		// delete messages from incomming queue
		synchronized (incommingMessages) {
			messageQueue = incommingMessages.get(connectionId);
		}
		if (messageQueue != null) {
			synchronized (messageQueue) {
				messageQueue.clear();
			}
		}
		messageQueue = null;

		// move messages from outcomming queue back to destination queue
		synchronized (outcommingMessages) {
			messageQueue = outcommingMessages.get(connectionId);
		}
		if (messageQueue != null) {
			synchronized (messageQueue) {
				for (Message message : messageQueue) {
					getDestinationManager(message.getJMSDestination()).put(message.getJMSDestination(), message);
				}
				messageQueue.clear();
			}
		}
	}
}
